package io.sodabox.mod;

import io.sodabox.common.utils.Amount;
import io.sodabox.common.utils.Time;
import io.sodabox.common.zookeeper.ZooKeeperClient;
import io.sodabox.common.zookeeper.ZooKeeperClient.Credentials;
import io.sodabox.common.zookeeper.ZooKeeperClient.ZooKeeperConnectionException;
import io.sodabox.common.zookeeper.ZooKeeperUtils;

import java.net.InetSocketAddress;
import java.util.List;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper.States;
import org.vertx.java.busmods.BusModBase;
import org.vertx.java.core.Handler;
import org.vertx.java.core.eventbus.Message;
import org.vertx.java.core.json.JsonObject;
import org.vertx.java.core.logging.Logger;

public class SodaboxModule extends BusModBase implements Handler<Message<JsonObject>> {

	private String 	address;

	private String 	host;
	private int 	port;
	private int 	timeout;

	private String	rootPath;

	private ZooKeeperClient zkClient;

	private Logger log;

	public void start() {
		super.start();

		log = container.getLogger();

		address = getOptionalStringConfig("address", "sodabox.sender");
		host = getOptionalStringConfig("host", "127.0.0.1");
		port = getOptionalIntConfig("port", 2181);
		timeout = getOptionalIntConfig("timeout", ZooKeeperUtils.DEFAULT_ZK_SESSION_TIMEOUT.getValue());

		rootPath = getOptionalStringConfig("rootPath", "/SODABOX/server_mq");

		zkClient = new ZooKeeperClient(
				Amount.of(timeout, Time.SECONDS), 
				Credentials.NONE,
				InetSocketAddress.createUnresolved(host, port)
				);

		try {
			initSenderModule();
		} catch (ZooKeeperConnectionException | InterruptedException
				| KeeperException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		eb.registerHandler(address, this);

	}

	private void initSenderModule() throws ZooKeeperConnectionException, InterruptedException, KeeperException{

		ZooKeeperUtils.ensurePath(zkClient, ZooDefs.Ids.OPEN_ACL_UNSAFE, rootPath);

		zkClient.get().getChildren(rootPath, new Watcher() {

			public void process(WatchedEvent event) {
				//System.err.println("Children Event: " + event);
				try {

					List<String> children = zkClient.get().getChildren(rootPath, this);
					System.out.println("Members: "+ event.getType()+" - " + children);

				} catch (Exception e) {
					throw new RuntimeException(e);
				}
			}

		});

		//Thread.sleep(4000);

		zkClient.get().create(rootPath+"/ABCDE", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);

		//Thread.sleep(4000);
		zkClient.get().create(rootPath+"/ABCDE123", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);

		//Thread.sleep(4000);
		zkClient.get().create(rootPath+"/ABCDE12311", new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);

	}

	public void stop() {
		try {
			super.stop();
		} catch (Exception e) {
		}
	}

	@Override
	public void handle(Message<JsonObject> message) {


		String action = message.body.getString("action");

		System.out.println(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>action : "+action);

		if (action == null) {
			sendError(message, "action must be specified");
			return;
		}

		switch (action) {

		case "status":
			status(message);
			break;
		default:
			sendError(message, "Invalid action: " + action);
			return;
		}


	}


	private void status(Message<JsonObject> message) {

		JsonObject reply = new JsonObject();

		try {
			if(null != zkClient && zkClient.get().getState().equals(States.CONNECTED) ){
				reply.putBoolean("connected", true);
			}else{
				reply.putBoolean("connected", false);
			}
		} catch (ZooKeeperConnectionException | InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
			reply.putBoolean("connected", false);
		}

		sendOK(message, reply);
	}


}
